package study.cdc.debezium.support.connector;

import io.debezium.data.Envelope;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.RecordChangeEvent;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.List;
import java.util.Map;

import static io.debezium.data.Envelope.FieldName.AFTER;
import static io.debezium.data.Envelope.FieldName.BEFORE;
import static io.debezium.data.Envelope.FieldName.OPERATION;
import static java.util.stream.Collectors.toMap;

@Slf4j
public class ChangeEvent_V2_Handler {

    public static void handlePayload(List<RecordChangeEvent<SourceRecord>> changeEvents,
                                     DebeziumEngine.RecordCommitter<RecordChangeEvent<SourceRecord>> recordCommitter) {
        //
        for (RecordChangeEvent<SourceRecord> changeEvent : changeEvents) {
            SourceRecord record = changeEvent.record();
            Struct value = (Struct) record.value();
            if (value == null) {
                continue;
            }
            String op = value.getString(OPERATION);
            log.info("operation= {}", op);
            Envelope.Operation operation = Envelope.Operation.forCode(op);
            String before = null, after = null;
            switch (operation) {
                case CREATE:
                    after = AFTER;
                    break;
                case DELETE:
                    before = BEFORE;
                    break;
                case UPDATE:
                    before = BEFORE;
                    after = AFTER;
                    break;
                case READ:
            }

            if (before != null) {
                Struct struct = (Struct) value.get(before);
                log.info("payload= {}", packageMap(struct));
            }
            if (after != null) {
                Struct struct = (Struct) value.get(after);
                log.info("payload= {}", packageMap(struct));
            }

//            String filed = operation == Envelope.Operation.DELETE ? BEFORE : AFTER;
//            log.info("filed= {}", filed);
//            Struct struct = (Struct) value.get(filed);
//            //将变更的行封装为 Map
//            Map<String, Object> payload = packageMap(struct);
//            log.info("payload= {}", payload);
        }
    }

    public static Map<String, Object> packageMap(Struct struct) {
        Map<String, Object> payload = struct.schema().fields().stream()
                .map(Field::name)
                .filter(fieldName -> struct.get(fieldName) != null)
                .map(fieldName -> Pair.of(fieldName, struct.get(fieldName)))
                .collect(toMap(Pair::getKey, Pair::getValue));
        return payload;
    }
}
